package com.lmq
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import org.apache.spark.sql.streaming.StreamingQuery


object WordCount {

  Logger.getLogger("org.apache.spark")
    .setLevel(Level.WARN)
  val spark = SparkSession
    .builder()
    .appName("StructuredNetworkWordCount")
    .master("local[*]")
    .getOrCreate()
  import spark.implicits._



  def main(args: Array[String]): Unit = {
    //Create DataFarame representing the stream of input lines from connection to localhost:9999
    val lines = spark.readStream
      .format("socket")
      .option("host", "localhost")
      .option("port", 9999)
      .load()

    // split the lines into words
    val words: Dataset[String] = lines.as[String].flatMap(_.split(" "))

    words.printSchema()


    // Generate running word count
    val wordsCount: DataFrame = words.groupBy("value").count()

    // Start running the query that prints the running counts to the console.
    val query: StreamingQuery = wordsCount.writeStream
      .outputMode("complete")
      .format("console")
      .start()



    query.awaitTermination()







  }

}
